package com.qqt.csr.im.service;

import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.qqt.csr.common.utils.CacheKeyUtil;
import com.qqt.csr.common.utils.JsonUtil;
import com.qqt.csr.im.entity.Account;
import com.qqt.csr.im.entity.Room;
import com.qqt.csr.im.entity.RoomMember;
import com.qqt.csr.im.mapper.RoomMemberMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
import org.redisson.api.BatchResult;
import org.redisson.api.RBatch;
import org.redisson.api.RFuture;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.amqp.RabbitTemplateConfigurer;
import org.springframework.stereotype.Service;

import java.time.Duration;
import java.time.temporal.ChronoUnit;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * im房间成员
 */
@Slf4j
@Service
public class RoomMemberService extends ServiceImpl<RoomMemberMapper, RoomMember> {
    /**
     * 本地缓存的房间成员，key:roomId
     */
    private static final Cache<Long, List<RoomMember>> ROOM_MEM_CACHE = Caffeine.newBuilder()
            .maximumSize(10000)
            .expireAfterWrite(10, TimeUnit.MINUTES)
            .build();
    @Autowired
    private RedissonClient redissonClient;

    /**
     * 新增房间成员
     * @param room
     * @param accountIdList
     */
    public void add(Room room, List<Long> accountIdList){
        if(CollectionUtils.isEmpty(accountIdList)){
            return;
        }
        List<RoomMember> exitMemList = queryMemberListBy(room.getId());
        List<Long> addAccountIdList = new ArrayList<>(accountIdList);
        if(CollectionUtils.isNotEmpty(exitMemList)){
            List<Long> exitMemIdList = exitMemList.stream().map(RoomMember::getAccountId).collect(Collectors.toList());
            addAccountIdList.removeIf(exitMemIdList::contains);
        }
        if(CollectionUtils.isEmpty(addAccountIdList)){
            return;
        }
        List<RoomMember> roomMemberList = addAccountIdList.stream()
                .map(accountId -> RoomMember.builder()
                        .accountId(accountId)
                        .roomId(room.getId())
                        .deleteFlag(0)
                        .build()
                ).collect(Collectors.toList());
        this.saveBatch(roomMemberList);
        room.setRoomMemberList(roomMemberList);
        clearCache(room.getId());
    }

    /**
     * 获取只包含集合中的成员共同的房间id
     * @param accountIdList
     * @return
     */
    public Long querySameRoomIdBy(List<Long> accountIdList){
        if(CollectionUtils.isEmpty(accountIdList)){
            return 0L;
        }
        return this.baseMapper.querySameRoomIdBy(accountIdList, accountIdList.size());
    }

    /**
     * 获取房间的所有成员
     * @param roomIdList
     * @return
     */
    public Map<Long, List<RoomMember>> queryMemberMapFromCacheBy(List<Long> roomIdList){
        if(CollectionUtils.isEmpty(roomIdList)){
            return Maps.newHashMap();
        }

        Map<Long, List<RoomMember>> roomMemMap = new HashMap<>();

        // 本地缓存获取
        List<Long> needQueryRedisRoomIdList = new ArrayList<>();
        roomIdList.forEach(roomId -> {
            List<RoomMember> roomMemberList = ROOM_MEM_CACHE.getIfPresent(roomId);
            if (CollectionUtils.isNotEmpty(roomMemberList)) {
                roomMemMap.put(roomId, roomMemberList);
            } else {
                needQueryRedisRoomIdList.add(roomId);
            }
        });

        // redis缓存获取
        Lists.partition(needQueryRedisRoomIdList, 100).forEach(pList -> {
            RBatch redisBatch = redissonClient.createBatch();
            pList.forEach(roomId -> {
                String redisKey = String.format(CacheKeyUtil.IM.ROOM_MEM_LIST_KEY, roomId);
                redisBatch.getBucket(redisKey).getAsync();
            });
            BatchResult<?> batchResult = redisBatch.execute();
            List<?> objectList = Optional.ofNullable(batchResult.getResponses()).orElseGet(Lists::newArrayList);
            objectList.forEach(obj -> {
                if (obj != null) {
                    Map<Long, List<RoomMember>> cacheResultMap = JsonUtil.toList(obj.toString(), RoomMember.class)
                            .stream().collect(Collectors.groupingBy(RoomMember::getRoomId));
                    roomMemMap.putAll(cacheResultMap);
                    ROOM_MEM_CACHE.putAll(cacheResultMap);
                }
            });
        });

        // 数据库查询
        List<Long> needQueryDbRoomIdList = needQueryRedisRoomIdList.stream()
                .filter(roomId -> !roomMemMap.containsKey(roomId)).collect(Collectors.toList());
        Map<Long, List<RoomMember>> dbRoomMemMap = queryMemberMapFromDbBy(needQueryDbRoomIdList);

        Lists.partition(needQueryDbRoomIdList, 100).forEach(pList -> {
            RBatch redisBatch = redissonClient.createBatch();
            pList.forEach(roomId -> {
                List<RoomMember> roomMemList = Optional.ofNullable(dbRoomMemMap.get(roomId)).orElseGet(Lists::newArrayList);
                roomMemMap.put(roomId, roomMemList);
                ROOM_MEM_CACHE.put(roomId, roomMemList);
                String redisKey = String.format(CacheKeyUtil.IM.ROOM_MEM_LIST_KEY, roomId);
                redisBatch.getBucket(redisKey).setAsync(JsonUtil.toJson(roomMemList), Duration.of(CacheKeyUtil.EXPIRE_TIME_8H, ChronoUnit.MILLIS));
            });
            redisBatch.execute();
        });

        return roomMemMap;
    }

    /**
     * 获取房间的所有成员
     * @param roomIdList
     * @return
     */
    public Map<Long, List<RoomMember>> queryMemberMapFromDbBy(List<Long> roomIdList){
        if(CollectionUtils.isEmpty(roomIdList)){
            return Maps.newHashMap();
        }
        return this.lambdaQuery().in(RoomMember::getRoomId, roomIdList)
                .eq(RoomMember::getDeleteFlag, 0)
                .list().stream().collect(Collectors.groupingBy(RoomMember::getRoomId));
    }

    /**
     * 获取房间的所有成员
     * @param roomId
     * @return
     */
    public List<RoomMember> queryMemberListBy(Long roomId){
        return Optional.ofNullable(queryMemberMapFromCacheBy(Lists.newArrayList(roomId)).get(roomId)).orElseGet(Lists::newArrayList);
    }

    private void clearCache(Long roomId) {
        ROOM_MEM_CACHE.invalidate(roomId);
        String redisKey = String.format(CacheKeyUtil.IM.ROOM_MEM_LIST_KEY, roomId);
        redissonClient.getBucket(redisKey).deleteAsync();
    }
}
